""""
Case Type   : D库常用语法兼容-常用函数兼容
Case Name   : gs_dump导入导出测试验证
Create At   : 2025-10-10
Owner       : chen-czywj
Description :
    1.使用兼容A库创建数据库
    2.创建插件
    3.执行测试
    4.执行备份
    5.清理环境
Expect      :
    1.成功
    2.成功
    3.成功
    4.成功
    5.清理环境
History     :
"""

import os
import unittest
from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from yat.test import Node
from yat.test import macro

log = Logger()
primary_sh = CommonSH('PrimaryDbUser')

class Function_Commonviews_Case0042(unittest.TestCase):
    def setUp(self):
        log.info(f'--------{os.path.basename(__file__)} start--------')
        self.constant = Constant()
        self.dbname_a = Node('PrimaryDbUser').db_name
        self.dbname = 'test_views_001'
        self.schema_name = 'sys_view_test_01'
        self.backup_path = os.path.join(os.path.dirname(macro.DB_INSTANCE_PATH),
                                        'dump_commonviews_042.tar')

    def test_Function_Commonviews(self):
        text = '-------------step1: 使用兼容D库创建数据库; expect: 成功-------------'
        log.info(text)
        msg = primary_sh.check_connection_status(self.constant.START_STATUS_MSG)
        self.assertTrue(msg)
        create_db = f'''drop database if exists {self.dbname};
            create database if not exists {self.dbname} dbcompatibility = 'D';'''
        res = primary_sh.execut_db_sql(create_db, dbname=self.dbname_a)
        log.info(res)
        self.assertIn(self.constant.CREATE_DATABASE_SUCCESS, res)

        text = '-------------step2: 创建插件; expect: 成功-------------'
        log.info(text)
        create_schema = f'''drop extension shark;
            create extension shark;
            create schema {self.schema_name};
            set search_path to {self.schema_name};'''
        res = primary_sh.execut_db_sql(create_schema, dbname=self.dbname)
        log.info(res)
        self.assertIn(self.constant.create_extension_success, res)
        self.assertIn(self.constant.CREATE_SCHEMA_SUCCESS_MSG, res)
        self.assertIn(self.constant.SET_SUCCESS_MSG, res)

        text = '-------------step3: 执行测试; expect: 成功-------------'
        log.info(text)
        create_tb = f"""set search_path to {self.schema_name};
        drop table if exists t_index;
create table t_index (id int primary key, c2 int not null, c3 char(1), 
    c4 text, c5 numeric(10, 2));
create unique index t_index_c2_uind on t_index(c2);
create index t_index_c3_c2_ind on t_index(c3, c2);
create index t_index_func_c4_ind on t_index(lower(c4));
create index t_index_hash_c3_ind on t_index using hash(c3);
create index t_index_filter_c5_ind on t_index (c5) where c5 > 100.00;
drop table if exists t_foreign1;
create table t_foreign1 (f_id int primary key, f_name text not null, 
    f_age int default 18);
drop table if exists t_foreign2;
create table t_foreign2 (f_id int primary key not null, 
    f_c2 int references t_foreign1(f_id), f_salary real check(f_salary > 0));
create table t_column (c1 int, c2 text, c3 char(1), 
    constraint t_stats_col_pk primary key (c1)) with (orientation = column);
create temp table t_temp_table as select * from t_index;
drop view if exists v_normal;
create view v_normal as select * from t_index;
drop view if exists v_check;
create view v_check as select * from t_index where id > 10 with check option;
drop materialized view if exists mv_normal;
create materialized view mv_normal as select * from t_foreign1;"""
        res = primary_sh.execut_db_sql(create_tb, dbname=self.dbname)
        log.info(res)
        self.assertIn(self.constant.SET_SUCCESS_MSG, res)
        self.assertIn(self.constant.CREATE_TABLE_SUCCESS, res)
        self.assertIn(self.constant.CREATE_INDEX_SUCCESS, res)
        self.assertIn('INSERT 0 0', res)
        self.assertIn(self.constant.CREATE_VIEW_SUCCESS_MSG, res)
        self.assertIn('CREATE MATERIALIZED VIEW', res)

        text = '------------------step4: 执行备份; expect: 成功------------------'
        log.info(text)
        result = primary_sh.exec_gs_dump(filename=self.backup_path, dbname=self.dbname)
        self.assertTrue(result, '执行失败:' + text)
        result = primary_sh.restore_file(self.backup_path,
                                              dbname=self.dbname)
        log.info(result)
        self.assertIn(self.constant.RESTORE_SUCCESS_MSG,
                      result, '执行失败' + text)

    def tearDown(self):
        text = '------------------step5: 清理环境; expect: 成功------------------'
        log.info(text)
        drop_db = f'''
            drop database if exists {self.dbname};'''
        res = primary_sh.execut_db_sql(drop_db, dbname=self.dbname_a)
        log.info(res)
        self.assertIn(self.constant.DROP_DATABASE_SUCCESS, res)
